package com.demo.config;

import com.demo.entity.MqFailInfo;
import com.demo.mapper.MqFailInfoMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import com.rabbitmq.client.Channel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Profile;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Date;


@Component
@Slf4j
@Profile("direct_queue_profiles")    // 方便不同环境测试
public class DirectReceiver {

    // 存入redis中的错误队列信息
    public static String MQ_ERROR_INFO = "mq_error_info";

    @Autowired
    private MqFailInfoMapper mqFailInfoMapper;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    int i = 0;

    @RabbitListener(queues = {
            "directQueue"
    })
    @RabbitHandler
    public void process(Message message, Channel channel) throws Exception {
        log.info("接收 directQueue 内容-->{}", message);
        if (!messageCheck(message, channel)) {  // 消息校验
            return;
        }
        try {

            // 模拟 消费成功或失败情况 -------------------begin
            double v = Math.random() * 10;
            if (v > 8) {
                // 业务逻辑，这里业务如果出错，会再次执行（spring.retry重复机制），注意：手动回滚业务
                businessCodeFail(message, channel);
            } else {
                // 成功
                businessCodeSuccess(message, channel);
                i++;
            }
            System.out.println("成功" + i);

            // 模拟 消费成功或失败情况 --------------------end
            // 手动确认消息
            messageSuccess(message, channel);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            dealFailMq(message, channel);
        }
    }


    /**
     * 业务代码-失败情况
     */
    public void businessCodeFail(Message message, Channel channel) {
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        throw new RuntimeException("我是失败业务测试");
    }

    /**
     * 业务代码-成功情况
     */
    public void businessCodeSuccess(Message message, Channel channel) {
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    /**
     * 校验消息是否重复操作。
     *
     * @param message
     * @param channel
     * @return
     * @throws IOException
     */
    public boolean messageCheck(Message message, Channel channel) throws IOException {
        String id = message.getMessageProperties().getMessageId();
        Object object = redisTemplate.opsForValue().get(id);
        if (object == null) {
            return true;
        } else {
            log.info("消息已经消费过了");
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return false;
        }
    }


    /**
     * 已消费消息放入redis，并手动确认消息
     *
     * @param message
     * @param channel
     * @throws IOException
     */
    public void messageSuccess(Message message, Channel channel) throws IOException {
        String id = message.getMessageProperties().getMessageId();
        redisTemplate.opsForValue().set(id, 1);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }


    /**
     * 错误消息
     *
     * @param message
     * @param channel
     * @throws IOException
     */
    public void dealFailMq(Message message, Channel channel) throws IOException {
        String id = message.getMessageProperties().getMessageId();
        String body = new String(message.getBody());

        mqFailInfoMapper.insert(new MqFailInfo() {
            {
                setMessageId(id);
                setBody(body);
                setExchangeName(message.getMessageProperties().getReceivedExchange());
                setRoutingKey(message.getMessageProperties().getReceivedRoutingKey());
                setCreateTime(new Date());
                setStatus(0);
            }
        });
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);

    }


}